package com.yonyou.distribute.zk;

import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.*;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;

/**
 * 实现原理：
 * znode节点不可以重复创建 且自旋
 * 但是自旋消耗性能
 * 解决:节点改为 临时序列化节点，监听比自己小的上一个节点
 */
public class ZKDistributedLock_2 implements Lock {

    private ZooKeeper zooKeeper;

    private String lockName;

    private String currentNodePath;

    private static final String ROOT_PATH = "/locks";

    public ZKDistributedLock_2(ZooKeeper zooKeeper, String lockName) {
        this.zooKeeper = zooKeeper;
        this.lockName = lockName;
        try {
            if(zooKeeper.exists(ROOT_PATH, null) == null){
               zooKeeper.create(ROOT_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void lock() {
        this.tryLock();
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    /**
     * 解决:节点改为 临时序列化节点，监听比自己小的上一个节点
     * @return
     */
    @Override
    public boolean tryLock() {
        // 创建节点,为了防止zk客户端获取到锁之后，服务器宕机死锁问题，这里创建临时节点
        try {
            // 所有请求获取锁时，给每一个请求创建一个临时节点
            currentNodePath = this.zooKeeper.create(ROOT_PATH + "/" + lockName + "-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 获取前一个节点，如果前一个节点为空，则获取锁成功，否则监听前一个节点
            String preNode = this.getPreNode();
            if(null != preNode){
                // 利用闭锁实现阻塞功能
                CountDownLatch countDownLatch = new CountDownLatch(1);
                // 因为获取前置节点这个操作不具备原子性，再次判断zk中是否存在前置节点
                if(this.zooKeeper.exists(ROOT_PATH + "/" + preNode, new Watcher() {
                    @Override
                    public void process(WatchedEvent watchedEvent) {
                        countDownLatch.countDown();
                    }
                }) == null){
                    return true;
                }
                countDownLatch.await();
            }
            return true;
        }catch (Exception e){

        }
        return false;
    }

    private String getPreNode() {
        //获取根节点下的所有子节点
        try {
            List<String> children = this.zooKeeper.getChildren(ROOT_PATH, false);
            if (CollectionUtils.isEmpty(children)) {
                throw new IllegalMonitorStateException("非法操作");
            }
            // 获取和当前节点同一资源的锁
            List<String> nodes = children.stream().filter(node -> StringUtils.startsWith(node, lockName + "-")).collect(Collectors.toList());
            if(CollectionUtils.isEmpty(nodes)){
                throw new IllegalMonitorStateException("非法操作");
            }
            // 排队
            Collections.sort(nodes);
            // 获取当前节点
            String currentNode = StringUtils.substringAfterLast(currentNodePath, "/");
            // 获取当前节点的下标
            int idx = Collections.binarySearch(nodes, currentNode);
            if(idx < 0){
                throw new IllegalMonitorStateException("非法操作");
            }else if(idx > 0){
                // 返回前一个节点
                return nodes.get(idx -1);
            }else {
                // 如果当前节点就是第一个节点
                return null;
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new IllegalMonitorStateException("非法操作");
        }
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }



    @Override
    public void unlock() {
        //解锁 删除节点
        try {
            zooKeeper.delete(currentNodePath, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }
}
